package com.xuliugen.storm.demo.trident.topology;

import com.xuliugen.storm.demo.trident.operator.filter.DiseaseFilter;
import com.xuliugen.storm.demo.trident.operator.function.CityAssignmentFunction;
import com.xuliugen.storm.demo.trident.operator.function.DispatchAlertFunction;
import com.xuliugen.storm.demo.trident.operator.function.HourAssignmentFunction;
import com.xuliugen.storm.demo.trident.operator.function.OutbreakDetectorFunction;
import com.xuliugen.storm.demo.trident.spout.DiagnosisEventSpout;
import com.xuliugen.storm.demo.trident.state.OutbreakTrendFactory;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.builtin.Count;
import org.apache.storm.tuple.Fields;

public class OutbreakDetectionTopology {

    public static StormTopology buildTopology() {

        TridentTopology topology = new TridentTopology();
        DiagnosisEventSpout spout = new DiagnosisEventSpout();
        Stream inputStream = topology.newStream("event", spout);

        /**
         * 应用运算需要声明一个输入域集合和一个输出域集合，也就是function域。
         * 在每个tuple中，CityAssignmentFunction会在event字段上运算并且增加一个叫做city的新字段，这个字段会附在tuple中向后发射。
         */
        inputStream.each(new Fields("event"), new DiseaseFilter()) //事件由DiseaseFilter函数过滤，过滤掉我们不关心的疾病事件
                .each(new Fields("event"), new CityAssignmentFunction(), new Fields("city"))  //事件由CityAssignment函数赋值一个对应的城市名
                .each(new Fields("event", "city"), new HourAssignmentFunction(), new Fields("hour", "cityDiseaseHour")) //然后HourAssignment函数赋值一个表示小时的时间戳，并且增加一个key cityDiseaseHour到tuple的字段中，这个key包括城市、小时和疾病代码
                /**
                 * 将特定字段值相同的tuple分组到同一个分片中
                 * 分片后aggregate函数在每个分片数据的每个分组中运行。
                 * cityDiseaseHour是根据城市、疾病代码、小时组合后的作为分组的关键词
                 *
                 * 返回的每一个组，然后Count aggregate在每个分组上执行，将计数发射给下游的消费者
                 */
                .groupBy(new Fields("cityDiseaseHour"))  //后续使用这个key进行分组统计并使用persistentAggregate函数对统计量持久性存储
                /**
                 * 使用的是内置的Count的Aggregator,经信息进行持久化存储
                 */
                .persistentAggregate(new OutbreakTrendFactory(), new Count(), new Fields("count")).newValuesStream()
                .each(new Fields("cityDiseaseHour", "count"), new OutbreakDetectorFunction(), new Fields("alert")) //统计量传递给OutbreakDetector函数，如果统计量超过阀值，OutbreakDetector向后发送一个告警信息
                .each(new Fields("alert"), new DispatchAlertFunction(), new Fields()); //最后DispatchAlert收到告警信息，记录日志，并且结束流程
        return topology.build();
    }

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("cdc", conf, buildTopology());
        Thread.sleep(600000 * 100);
        cluster.shutdown();
    }
}
